package com.shujia.ads

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
import org.apache.flink.table.catalog.hive.HiveCatalog

object Demo1ComputeGenderIndex {
  def main(args: Array[String]): Unit = {



    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() //使用blink计划器
      .inStreamingMode() //流处理模型
      .build()


    //创建table环境
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    val hiveCatalog: HiveCatalog = new HiveCatalog("myHive","sent","sentcompute/src/main/resources")

    bsTableEnv.registerCatalog("myHive",hiveCatalog)

    bsTableEnv.useCatalog("myHive")



    bsTableEnv.executeSql(
      """
        |
        |insert into ads.ads_mysql_gender_index
        |select b.gender,count(comment_id) as c from
        |dwd.dwd_kafka_comment_msk as a inner join dwd.dwd_kafka_user_msk as b
        |on a.user_id=b.id
        |group by b.gender
        |
        |
      """.stripMargin)



  }
}
